package com.xuliugen.storm.demo.simple.bolts;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.io.FileWriter;
import java.io.IOException;
import java.util.Map;
import java.util.UUID;

/**
 * 为每一个名字加上后缀
 * @author xuliugen
 */
public class AppendBolt extends BaseRichBolt {

    private static final long serialVersionUID = 1L;

    FileWriter fileWriter = null;

    @Override
    public void execute(Tuple tuple) {
        // 先拿到上一个组件发送过来的姓名
        String upperName = tuple.getString(0);
        String suffix_name = upperName + "_csdn";
        try {
            fileWriter.write(suffix_name);
            fileWriter.write("\n");
            fileWriter.flush();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 在bolt组件运行过程中只会被调用一次
     */
    @Override
    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
        try {
            fileWriter = new FileWriter("/home/xuliugen/server/stormdata/" + UUID.randomUUID());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 本bolt已经不需要发送tuple消息到下一个组件，所以不需要再声明tuple的字段
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer arg0) {
    }

}
